package com.atguigu;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * 什么时候用sql，什么时候需要写代码
 * 1.对接外部存储
 * 2.动态拼接sql
 * 3.要用到第三方工具类
 */
public class Hbase_DML {
    //获取到hbase链接
    public static Connection connection = HbaseConnectionSingle.connection;

    //插入数据
    public static void putCell(String namespace, String tableName, String rowKey, String family, String column, String value) throws IOException {
        //1.通过Connection获取Table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        //2.创建Put对象
        Put put = new Put(Bytes.toBytes(rowKey));

        //3.指定要往哪个列族，哪个列插入什么数据
        put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));

        //4.调用put写入数据
        table.put(put);

        //关闭Table链接
        table.close();
    }

    //通过get获取某一行数据
    public static void getCells(String namespace, String tableName, String rowKey, String family, String column) throws IOException {
        //1.通过Connection获取Table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        //2.创建Get对象
        Get get = new Get(Bytes.toBytes(rowKey));
        //指定获取一行中那一列的数据 (在获取数据的时候就筛选出某一列)  where
//        get.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));

        //指定获取一行中那一列族的数据
        get.addFamily(Bytes.toBytes(family));

        //获取所有版本的数据
        get.readAllVersions();

        //获取执行版本个数的数据
        //get.readVersions(8);


        //调用get方法获取数据
        Result result = table.get(get);

        // byte[] value = result.value();

        //指定获取一行中那一列的数据（根据已经获取到的一行数据再筛选出某一列） having
                byte[] value = result.getValue(Bytes.toBytes(family), Bytes.toBytes(column));

        //从返回结果中获取一行的数据
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.println(Bytes.toString(CellUtil.cloneRow(cell)) + ":" + Bytes.toString(CellUtil.cloneFamily(cell)) + ":" + Bytes.toString(CellUtil.cloneQualifier(cell)) + ":" + Bytes.toString(CellUtil.cloneValue(cell)));
        }

//        System.out.println(Bytes.toString(value));


        //关闭Table资源
        table.close();
    }

    //删除数据
    public static void deleteCells(String namespace, String tableName, String rowKey, String family, String column) throws IOException {
     //1.通过Connection获取Table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        //2.创建Delete对象
        Delete delete = new Delete(Bytes.toBytes(rowKey));

        //删除某一列的数据
//        delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));

        //删除指定列所有版本的数据
        delete.addColumns(Bytes.toBytes(family), Bytes.toBytes(column));


        //3.调用delete方法
        table.delete(delete);

        //关闭Table
        table.close();
    }
    //扫描数据
    public static void scanRows(String namespace, String tableName, String startRow ,String stopRow) throws IOException {
        //1.通过Connection获取Table对象
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        //2.创建Scan对象
        Scan scan = new Scan();


        //设置扫描范围
//        scan.setStartRow();
        //默认情况下包含开始行不包含结束行  如果想要设置是否包含开始或结束行，需要指定第二个参数，true就是包含 false不包含
        scan.withStartRow(Bytes.toBytes(startRow));
        scan.withStopRow(Bytes.toBytes(stopRow),true);

        //获取超过版本的历史数据
        scan.setRaw(true);

        //获取所有版本的数据
        scan.readAllVersions();

        //获取指定版本的数据
//        scan.readVersions(6);


        ResultScanner scanner = table.getScanner(scan);

        //遍历ResultScanner获取Result然后再获取Cell
        for (Result result : scanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                System.out.println(Bytes.toString(CellUtil.cloneRow(cell)) + ":" + Bytes.toString(CellUtil.cloneFamily(cell)) + ":" + Bytes.toString(CellUtil.cloneQualifier(cell)) + ":" + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        //关闭Table
        table.close();
    }



    public static void main(String[] args) throws IOException {

//        putCell("bigdata", "student", "1001", "info", "name", "zhaoyi");
//        putCell("bigdata", "student", "1001", "info", "age", "18");
//        putCell("bigdata", "student", "1003", "info", "name", "zsf");
//        putCell("bigdata", "student", "1003", "info", "age", "19");
//        putCell("bigdata", "student", "1004", "info", "name", "ls");
//        putCell("bigdata", "student", "1004", "info", "age", "20");
//        putCell("bigdata", "student", "1005", "info", "name", "wangwu");
//        putCell("bigdata", "student", "1005", "info", "age", "21");
//        putCell("bigdata", "student", "1006", "info", "name", "sunliu");
//        putCell("bigdata", "student", "1006", "info", "age", "22");
//
//        putCell("bigdata", "student", "1002", "info", "name", "zs");
//        putCell("bigdata", "student", "1002", "info", "age", "18");
//        putCell("bigdata", "student", "1002", "info", "age", "19");
//        putCell("bigdata", "student", "1002", "info", "age", "20");
//        putCell("bigdata", "student", "1002", "info", "age", "21");
//        putCell("bigdata", "student", "1002", "info", "age", "22");
//        putCell("bigdata", "student", "1002", "info", "age", "23");
//        putCell("bigdata", "student", "1002", "info", "age", "24");

//        getCells("bigdata", "student", "1002", "info", "age");

//        deleteCells("bigdata", "student", "1002", "info", "age");

        scanRows("bigdata", "student", "1001","1004");

        HbaseConnectionSingle.closeConnection();

    }
}
